package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo05EventTime {
    public static void main(String[] args) throws Exception {
        // 事件时间：在数据中存在一列，记录着数据产生的时间


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 当并行度不为1时，Task的数量会大于1的，所以相当于有多个Task在并行执行
         * 所以Task与Task之间也是需要进行水位线对齐的
         *
         * 将并行度设为1，那就不会存在Task之间水位线对不齐的问题，因为只有一个Task
         *
         * 在真实的场景中，由于数据量是远超过并行度的，所以Task之间对其水位线是相对更容易的
         */
        env.setParallelism(1);

        DataStream<String> wordDs = env.socketTextStream("master", 8888);

        DataStream<Tuple2<String, Long>> wordTimeDS = wordDs.map(line -> {
            String[] splits = line.split(",");
            String word = splits[0];
            long ts = Long.parseLong(splits[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        DataStream<Tuple2<String, Long>> assDS = wordTimeDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                // 1、使用数据中最大的时间作为水位线
//                                .<Tuple2<String, Long>>forMonotonousTimestamps()
                                // 2、将水位线前移5s，解决数据延时到达的问题
                                // 前移的太多就会导致整体任务延时较大
                                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                // 告诉flink，数据中哪部分作为时间戳，即使用事件时间
                                .withTimestampAssigner(TimestampAssignerSupplier.of((kv, ts) -> kv.f1)
                                ));

        DataStream<Tuple2<String, Integer>> kvDS = assDS.map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));

        // 基于事件时间统计5s内的单词数量
        kvDS.keyBy(kv -> kv.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();

    }
}
